package com.linkzd.iot.utils.mqtt.client;

import com.linkzd.common.utils.StringUtils;
import com.linkzd.iot.domain.DeviceInfo;
import com.linkzd.iot.domain.DeviceFeedbackModel;
import com.linkzd.iot.service.IDeviceFeedbackService;
import com.linkzd.iot.utils.mqtt.model.QosModel;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @author: Wangym
 * @Date: 2024/5/24 16:31
 * @Description:
 */
@Component
@Slf4j
public class MqttClientUtil {
    @Value("${spring.mqtt.username}")
    private String username;
    @Value("${spring.mqtt.password}")
    private String password;
    @Value("${spring.mqtt.hostUrl}")
    private String host;
    @Value("${spring.mqtt.default-topic}")
    private String topic;
    @Value("${spring.mqtt.client-id}")
    private String clientId;
    @Value("${spring.mqtt.timeout}")
    private int timeout;
    @Value("${spring.mqtt.keepalive}")
    private int interval;
    @Value("${spring.mqtt.device-feedback-time-out}")
    private int deviceFeedbackTimeOut;

//    @Value("${spring.mqtt.auto-connect}")
//    private boolean autoConnect;

    @Autowired
    private ParsingData parsingData;

    private MqttClient mqttClient;
    private MqttConnectOptions mqttConnectOptions;

    @Autowired
    private IDeviceFeedbackService deviceFeedbackService;


    @PostConstruct
    private void init() {
        System.out.println("" +
                "MQTT服务启动成功" +
                "                                        \n" +
                ",--.   ,--.  ,-----.    ,--------. ,--------. \n" +
                "|   `.'   | '  .-.  '   '--.  .--' '--.  .--' \n" +
                "|  |'.'|  | |  | |  |      |  |       |  |    \n" +
                "|  |   |  | '  '-'  '-.    |  |       |  |    \n" +
                "`--'   `--'  `-----'--'    `--'       `--'    \n" +
                "                                              ");
        connect(host, clientId);
        //如果启动MQTT服务
        //将多个MQTT主题用逗号分隔
        String mqtt_topic[] = StringUtils.split(topic, ",");
        //连接MATT服务器
        //遍历数组中的每一个主题
        for (int i = 0; i < mqtt_topic.length; i++) {
            //订阅主题
            try {
                subscribe(mqtt_topic[i], QosModel.CONNECTION_METHOD_1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 连接mqtt
     *
     * @param host
     * @param clientId
     */
    public void connect(String host, String clientId) {
        try {
            mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
            mqttConnectOptions = getMqttConnectOptions();
            mqttClient.connect(mqttConnectOptions);
        } catch (Exception e) {
            System.out.println("mqtt服务连接异常!");
            e.printStackTrace();
        }
    }


    /**
     * 设置连接对象信息
     * setCleanSession  true 断开连接即清楚会话  false 保留连接信息 离线还会继续发消息
     *
     * @return
     */
    private MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{host});
        mqttConnectOptions.setKeepAliveInterval(interval);
        mqttConnectOptions.setConnectionTimeout(timeout);
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setAutomaticReconnect(false);
        return mqttConnectOptions;
    }

    /**
     * mqtt连接状态
     *
     * @return
     */
    private boolean isConnect() {
        if (Objects.isNull(this.mqttClient)) {
            return false;
        }
        return mqttClient.isConnected();
    }

    /**
     * 设置重连
     *
     * @throws Exception
     */
    private void reConnect() throws Exception {
        if (Objects.nonNull(this.mqttClient)) {
            System.out.println("mqtt 服务已重新连接...");
            this.mqttClient.connect(this.mqttConnectOptions);
        }
    }

    /**
     * 断开连接
     *
     * @throws Exception
     */
    public void closeConnect() throws Exception {
        if (Objects.nonNull(this.mqttClient)) {
            System.out.println("mqtt 服务已断开连接...");
            this.mqttClient.disconnect();
        }
    }


    /** @Author Wangym
     * @Date 2024/5/26 上午 12:05
     * @Description 获取设备反馈消息数据库中的设备
     * @Param
     * @Return
     **/
    private Future<Boolean> getDeviceFeedbackData(String code){
        Callable<Boolean> callableTask = () -> {
            int count = 0;
            while (count++<deviceFeedbackTimeOut){
                List<DeviceFeedbackModel> deviceFeedbackModels = deviceFeedbackService.selectDeviceFeedbackList(code);
                if(deviceFeedbackModels.size() > 0){
                    int rows = deviceFeedbackService.updateDeviceFeedback(code);
                    if(rows > 0){
                        return true;
                    }
                }
                // 模拟长时间的操作
                Thread.sleep(1000);
            }
            return false;
        };
        // 提交任务并获取Future对象
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        return executorService.submit(callableTask);
    }


    /** @Author Wangym
     * @Date 2024/5/25 下午 10:49
     * @Description 设备等待反馈机制
     * @Param
     * @Return
     **/

    private Boolean deviceFeedback(String code){
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<Boolean> future = this.getDeviceFeedbackData(code);

        try {
            // 设置最大等待时间，这里设置为5秒
            Boolean result = future.get(deviceFeedbackTimeOut, TimeUnit.SECONDS);
            if (result) {
                log.info("设备已反馈成功信息!");
                return true;
            } else {
                log.info("等待设备反馈超时!");
            }
        } catch (Exception e) {
            log.info("等待设备反馈超时!");
            // 取消任务
            future.cancel(true);
        }finally {
            // 关闭线程池
            executor.shutdownNow();
        }
        return false;
    }
        /** @Author Wangym
     *  retained    是否保留 true
     * @Date 2024/5/15 上午 8:52
     * @Description (默认调用)安全等级中
     * @Param
     * @Return
     **/
    public String publishMessage(String topic, DeviceInfo deviceInfo){
        if (deviceInfo == null) {
            return "设备信息为空!";
        }
        if(deviceInfo.getDeviceCode() == null || deviceInfo.getDeviceCode() == ""){
            return "设备编码为空!";
        }
        if(deviceInfo.getDeviceData() == null || deviceInfo.getDeviceData() == ""){
            return "设备数据为空!";
        }
        String pushMessage = "exe(" + deviceInfo.getDeviceCode() + "," + deviceInfo.getDeviceData() + ")\r\n";
        String result = null;
        try {
            Boolean b = this.sendMessage(topic,pushMessage, QosModel.CONNECTION_METHOD_1,deviceInfo.getDeviceCode());
            if(b){
                result = "true";
            }else{
                result = "等待设备反馈超时!";
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }


    /**
     * 发布消息
     *
     * @param topic
     * @param message
     * @param qos
     * @throws Exception
     */
    public Boolean sendMessage(String topic, String message, int qos,String code) throws Exception {
        if (Objects.nonNull(this.mqttClient) && this.mqttClient.isConnected()) {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setPayload(message.getBytes());
            mqttMessage.setQos(qos);

            MqttTopic mqttTopic = mqttClient.getTopic(topic);

            if (Objects.nonNull(mqttTopic)) {
                MqttDeliveryToken publish = null;
                try {
                    publish = mqttTopic.publish(mqttMessage);
                    log.info("设备数据下发{设备编号:" + code + ";报文:" + message + "}");
                    return this.deviceFeedback(code);//等待设备反馈
                } catch (MqttPersistenceException e) {
                    e.printStackTrace();
                    System.out.println("消息发送异常!");
                } catch (MqttException e) {
                    e.printStackTrace();
                    System.out.println("消息发送异常!");
                }

            }
        } else {
            this.reConnect();
        }
        return false;
    }

    /**
     * 订阅消息
     *
     * @param topic
     * @param qos
     * @throws Exception
     */
    public void subscribe(String topic, int qos) throws Exception {
        if (Objects.nonNull(this.mqttClient) && this.mqttClient.isConnected()) {
            mqttClient.setCallback(new MqttCallback() {
                //连接丢失处理
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("连接已丢失");
                    cause.printStackTrace();
                    try {
                        connect(host, clientId);
                        subscribe(topic, QosModel.CONNECTION_METHOD_1);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

                //消息处理
                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    log.info("接收消息Topic : " + topic);
                    log.info("接收消息Qos : " + mqttMessage.getQos());

                    String message =new String(mqttMessage.getPayload());
                    if(message.split(",").length>1){
                        log.info("接收消息Message : " + message);
                        String[] messages = message.split(",");
                        if("ok\r\n".equals(messages[1])){
                            //解析执行器反馈信息
                            parsingData.feedback(messages);
                        }else{
                            //解析传感器主动上报数据
                            parsingData.analyze(messages);
                        }
                    }else{
                        log.error("接收消息格式异常!");
                    }
                    
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("消息发送已完成");
                }
            });
            mqttClient.subscribe(topic, qos);
        } else {
            reConnect();
        }
    }
}
